/*
 * dblink.c
 *
 * 从远程数据库返回结果的函数
 *
 * Joe Conway <mail@joeconway.com>
 * 及贡献者：
 * Darko Prenosil <Darko.Prenosil@finteh.hr>
 * Shridhar Daithankar <shridhar_daithankar@persistent.co.in>
 *
 * contrib/dblink/dblink.c
 * Copyright (c) 2001-2022, PostgreSQL Global Development Group
 * ALL RIGHTS RESERVED;
 *
 * Permission to use, copy, modify, and distribute this software and its
 * documentation for any purpose, without fee, and without a written agreement
 * is hereby granted, provided that the above copyright notice and this
 * paragraph and the following two paragraphs appear in all copies.
 *
 * IN NO EVENT SHALL THE AUTHOR OR DISTRIBUTORS BE LIABLE TO ANY PARTY FOR
 * DIRECT, INDIRECT, SPECIAL, INCIDENTAL, OR CONSEQUENTIAL DAMAGES, INCLUDING
 * LOST PROFITS, ARISING OUT OF THE USE OF THIS SOFTWARE AND ITS
 * DOCUMENTATION, EVEN IF THE AUTHOR OR DISTRIBUTORS HAVE BEEN ADVISED OF THE
 * POSSIBILITY OF SUCH DAMAGE.
 *
 * THE AUTHOR AND DISTRIBUTORS SPECIFICALLY DISCLAIMS ANY WARRANTIES,
 * INCLUDING, BUT NOT LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY
 * AND FITNESS FOR A PARTICULAR PURPOSE.  THE SOFTWARE PROVIDED HEREUNDER IS
 * ON AN "AS IS" BASIS, AND THE AUTHOR AND DISTRIBUTORS HAS NO OBLIGATIONS TO
 * PROVIDE MAINTENANCE, SUPPORT, UPDATES, ENHANCEMENTS, OR MODIFICATIONS.
 *
 */
#include "postgres.h"

#include <limits.h>

#include "access/htup_details.h"
#include "access/relation.h"
#include "access/reloptions.h"
#include "access/table.h"
#include "catalog/namespace.h"
#include "catalog/pg_foreign_data_wrapper.h"
#include "catalog/pg_foreign_server.h"
#include "catalog/pg_type.h"
#include "catalog/pg_user_mapping.h"
#include "executor/spi.h"
#include "foreign/foreign.h"
#include "funcapi.h"
#include "lib/stringinfo.h"
#include "libpq-fe.h"
#include "mb/pg_wchar.h"
#include "miscadmin.h"
#include "parser/scansup.h"
#include "utils/acl.h"
#include "utils/builtins.h"
#include "utils/fmgroids.h"
#include "utils/guc.h"
#include "utils/lsyscache.h"
#include "utils/memutils.h"
#include "utils/rel.h"
#include "utils/varlena.h"

PG_MODULE_MAGIC;

typedef struct remoteConn
{
	PGconn	   *conn;			/* 保持远程连接 */
	int			openCursorCount;	/* 打开的游标数量 */
	bool		newXactForCursor;	/* 为游标打开了一个事务 */
} remoteConn;

typedef struct storeInfo
{
	FunctionCallInfo fcinfo;
	Tuplestorestate *tuplestore;
	AttInMetadata *attinmeta;
	MemoryContext tmpcontext;
	char	  **cstrs;
	/* 临时存储结果以避免在异常时泄漏 */
	PGresult   *last_res;
	PGresult   *cur_res;
} storeInfo;

/*
 * 内部声明
 */
static Datum fc_dblink_record_internal(FunctionCallInfo fc_fcinfo, bool fc_is_async);
static void fc_prepTuplestoreResult(FunctionCallInfo fc_fcinfo);
static void fc_materializeResult(FunctionCallInfo fc_fcinfo, PGconn *fc_conn,
							  PGresult *fc_res);
static void fc_materializeQueryResult(FunctionCallInfo fc_fcinfo,
								   PGconn *fc_conn,
								   const char *fc_conname,
								   const char *fc_sql,
								   bool fc_fail);
static PGresult *fc_storeQueryResult(volatile storeInfo *fc_sinfo, PGconn *fc_conn, const char *fc_sql);
static void fc_storeRow(volatile storeInfo *fc_sinfo, PGresult *fc_res, bool fc_first);
static remoteConn *fc_getConnectionByName(const char *fc_name);
static HTAB *fc_createConnHash(void);
static void fc_createNewConnection(const char *fc_name, remoteConn *fc_rconn);
static void fc_deleteConnection(const char *fc_name);
static char **fc_get_pkey_attnames(Relation fc_rel, int16 *fc_indnkeyatts);
static char **fc_get_text_array_contents(ArrayType *fc_array, int *fc_numitems);
static char *fc_get_sql_insert(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals, char **fc_tgt_pkattvals);
static char *fc_get_sql_delete(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_tgt_pkattvals);
static char *fc_get_sql_update(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals, char **fc_tgt_pkattvals);
static char *fc_quote_ident_cstr(char *fc_rawstr);
static int	fc_get_attnum_pk_pos(int *fc_pkattnums, int fc_pknumatts, int fc_key);
static HeapTuple fc_get_tuple_of_interest(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals);
static Relation fc_get_rel_from_relname(text *fc_relname_text, LOCKMODE fc_lockmode, AclMode fc_aclmode);
static char *fc_generate_relation_name(Relation fc_rel);
static void fc_dblink_connstr_check(const char *fc_connstr);
static void fc_dblink_security_check(PGconn *fc_conn, remoteConn *fc_rconn);
static void fc_dblink_res_error(PGconn *fc_conn, const char *fc_conname, PGresult *fc_res,
							 bool fc_fail, const char *fc_fmt,...) pg_attribute_printf(5, 6);
static char *fc_get_connect_string(const char *fc_servername);
static char *fc_escape_param_str(const char *fc_from);
static void fc_validate_pkattnums(Relation fc_rel,
							   int2vector *fc_pkattnums_arg, int32 fc_pknumatts_arg,
							   int **fc_pkattnums, int *fc_pknumatts);
static bool fc_is_valid_dblink_option(const PQconninfoOption *fc_options,
								   const char *fc_option, Oid fc_context);
static int	fc_applyRemoteGucs(PGconn *fc_conn);
static void fc_restoreLocalGucs(int fc_nestlevel);

/* 全局 */
static remoteConn *pconn = NULL;
static HTAB *remoteConnHash = NULL;

/*
 *	以下是包含多个远程连接的列表。
 *	每个 dblink 函数的调用约定更改为接受
 *	连接名称作为第一个参数。连接列表
 *	与 ecpg 很像，例如名称与 PGconn 对象之间的映射。
 */

typedef struct remoteConnHashEnt
{
	char		name[NAMEDATALEN];
	remoteConn *rconn;
} remoteConnHashEnt;

/* 连接哈希的初始数量 */
#define NUMCONN 16

static char * fc_xpstrdup(const char *fc_in)
{
	if (fc_in == NULL)
		return NULL;
	return pstrdup(fc_in);
}

static void pg_attribute_noreturn()
fc_dblink_res_internalerror(PGconn *fc_conn, PGresult *fc_res, const char *fc_p2)
{
	char	   *fc_msg = pchomp(PQerrorMessage(fc_conn));

	if (fc_res)
		PQclear(fc_res);
	elog(ERROR, "%s: %s", fc_p2, fc_msg);
}

static void pg_attribute_noreturn()
fc_dblink_conn_not_avail(const char *fc_conname)
{
	if (fc_conname)
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
				 errmsg("connection \"%s\" not available", fc_conname)));
	else
		ereport(ERROR,
				(errcode(ERRCODE_CONNECTION_DOES_NOT_EXIST),
				 errmsg("connection not available")));
}

static void fc_dblink_get_conn(char *fc_conname_or_str,
				PGconn *volatile *fc_conn_p, char **fc_conname_p, volatile bool *fc_freeconn_p)
{
	remoteConn *fc_rconn = fc_getConnectionByName(fc_conname_or_str);
	PGconn	   *fc_conn;
	char	   *fc_conname;
	bool		fc_freeconn;

	if (fc_rconn)
	{
		fc_conn = fc_rconn->conn;
		fc_conname = fc_conname_or_str;
		fc_freeconn = false;
	}
	else
	{
		const char *fc_connstr;

		fc_connstr = fc_get_connect_string(fc_conname_or_str);
		if (fc_connstr == NULL)
			fc_connstr = fc_conname_or_str;
		fc_dblink_connstr_check(fc_connstr);

		/*
		 * 我们必须遵守 fd.c 对非虚拟文件描述符的限制。假设
		 * 一个 PGconn 代表一个长时间存在的 FD。 （在此处执行此操作还
		 * 确保在需要时关闭 VFD。）
		 */
		if (!AcquireExternalFD())
		{
#ifndef WIN32					/* 不能在 ereport() 宏中写 #if */
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not establish connection"),
					 errdetail("There are too many open files on the local server."),
					 errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
#else
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not establish connection"),
					 errdetail("There are too many open files on the local server."),
					 errhint("Raise the server's max_files_per_process setting.")));
#endif
		}

		/* 可以建立连接 */
		fc_conn = PQconnectdb(fc_connstr);

		if (PQstatus(fc_conn) == CONNECTION_BAD)
		{
			char	   *fc_msg = pchomp(PQerrorMessage(fc_conn));

			PQfinish(fc_conn);
			ReleaseExternalFD();
			ereport(ERROR,
					(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
					 errmsg("could not establish connection"),
					 errdetail_internal("%s", fc_msg)));
		}
		fc_dblink_security_check(fc_conn, fc_rconn);
		if (PQclientEncoding(fc_conn) != GetDatabaseEncoding())
			PQsetClientEncoding(fc_conn, GetDatabaseEncodingName());
		fc_freeconn = true;
		fc_conname = NULL;
	}

	*fc_conn_p = fc_conn;
	*fc_conname_p = fc_conname;
	*fc_freeconn_p = fc_freeconn;
}

static PGconn * fc_dblink_get_named_conn(const char *fc_conname)
{
	remoteConn *fc_rconn = fc_getConnectionByName(fc_conname);

	if (fc_rconn)
		return fc_rconn->conn;

	fc_dblink_conn_not_avail(fc_conname);
	return NULL;				/* 保持编译器安静 */
}

static void fc_dblink_init(void)
{
	if (!pconn)
	{
		pconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext, sizeof(remoteConn));
		pconn->conn = NULL;
		pconn->openCursorCount = 0;
		pconn->newXactForCursor = false;
	}
}

/*
 * 创建与另一个数据库的持久连接
 */
PG_FUNCTION_INFO_V1(dblink_connect);
Datum dblink_connect(PG_FUNCTION_ARGS)
{
	char	   *fc_conname_or_str = NULL;
	char	   *fc_connstr = NULL;
	char	   *fc_connname = NULL;
	char	   *fc_msg;
	PGconn	   *fc_conn = NULL;
	remoteConn *fc_rconn = NULL;

	fc_dblink_init();

	if (PG_NARGS() == 2)
	{
		fc_conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(1));
		fc_connname = text_to_cstring(PG_GETARG_TEXT_PP(0));
	}
	else if (PG_NARGS() == 1)
		fc_conname_or_str = text_to_cstring(PG_GETARG_TEXT_PP(0));

	if (fc_connname)
	{
		fc_rconn = (remoteConn *) MemoryContextAlloc(TopMemoryContext,
												  sizeof(remoteConn));
		fc_rconn->conn = NULL;
		fc_rconn->openCursorCount = 0;
		fc_rconn->newXactForCursor = false;
	}

	/* 首先检查有效的外部数据服务器 */
	fc_connstr = fc_get_connect_string(fc_conname_or_str);
	if (fc_connstr == NULL)
		fc_connstr = fc_conname_or_str;

	/* 如果不是超用户，则检查连接字符串中的密码 */
	fc_dblink_connstr_check(fc_connstr);

	/*
	 * 我们必须遵守 fd.c 对非虚拟文件描述符的限制。假设
	 * 一个 PGconn 代表一个长时间存在的 FD。 （在此处执行此操作还
	 * 确保在需要时关闭 VFD。）
	 */
	if (!AcquireExternalFD())
	{
#ifndef WIN32					/* 不能在 ereport() 宏中写 #if */
		ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("could not establish connection"),
				 errdetail("There are too many open files on the local server."),
				 errhint("Raise the server's max_files_per_process and/or \"ulimit -n\" limits.")));
#else
		ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("could not establish connection"),
				 errdetail("There are too many open files on the local server."),
				 errhint("Raise the server's max_files_per_process setting.")));
#endif
	}

	/* 可以建立连接 */
	fc_conn = PQconnectdb(fc_connstr);

	if (PQstatus(fc_conn) == CONNECTION_BAD)
	{
		fc_msg = pchomp(PQerrorMessage(fc_conn));
		PQfinish(fc_conn);
		ReleaseExternalFD();
		if (fc_rconn)
			pfree(fc_rconn);

		ereport(ERROR,
				(errcode(ERRCODE_SQLCLIENT_UNABLE_TO_ESTABLISH_SQLCONNECTION),
				 errmsg("could not establish connection"),
				 errdetail_internal("%s", fc_msg)));
	}

	/* 如果不是超用户，检查实际使用的密码 */
	fc_dblink_security_check(fc_conn, fc_rconn);

	/* 如果需要，尝试设置客户端编码以匹配服务器编码 */
	if (PQclientEncoding(fc_conn) != GetDatabaseEncoding())
		PQsetClientEncoding(fc_conn, GetDatabaseEncodingName());

	if (fc_connname)
	{
		fc_rconn->conn = fc_conn;
		fc_createNewConnection(fc_connname, fc_rconn);
	}
	else
	{
		if (pconn->conn)
		{
			PQfinish(pconn->conn);
			ReleaseExternalFD();
		}
		pconn->conn = fc_conn;
	}

	PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

/*
 * 清除与另一个数据库的持久连接
 */
PG_FUNCTION_INFO_V1(dblink_disconnect);
Datum dblink_disconnect(PG_FUNCTION_ARGS)
{
	char	   *fc_conname = NULL;
	remoteConn *fc_rconn = NULL;
	PGconn	   *fc_conn = NULL;

	fc_dblink_init();

	if (PG_NARGS() == 1)
	{
		fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_rconn = fc_getConnectionByName(fc_conname);
		if (fc_rconn)
			fc_conn = fc_rconn->conn;
	}
	else
		fc_conn = pconn->conn;

	if (!fc_conn)
		fc_dblink_conn_not_avail(fc_conname);

	PQfinish(fc_conn);
	ReleaseExternalFD();
	if (fc_rconn)
	{
		fc_deleteConnection(fc_conname);
		pfree(fc_rconn);
	}
	else
		pconn->conn = NULL;

	PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

/*
 * 使用持久连接打开游标
 */
PG_FUNCTION_INFO_V1(dblink_open);
Datum dblink_open(PG_FUNCTION_ARGS)
{
	PGresult   *fc_res = NULL;
	PGconn	   *fc_conn;
	char	   *fc_curname = NULL;
	char	   *fc_sql = NULL;
	char	   *fc_conname = NULL;
	StringInfoData fc_buf;
	remoteConn *fc_rconn = NULL;
	bool		fc_fail = true;	/* 默认向后兼容的行为 */

	fc_dblink_init();
	initStringInfo(&fc_buf);

	if (PG_NARGS() == 2)
	{
		/* 文本，文本 */
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
		fc_rconn = pconn;
	}
	else if (PG_NARGS() == 3)
	{
		/* 可能是文本，文本，文本或文本，文本，布尔 */
		if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
		{
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
			fc_fail = PG_GETARG_BOOL(2);
			fc_rconn = pconn;
		}
		else
		{
			fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
			fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
			fc_rconn = fc_getConnectionByName(fc_conname);
		}
	}
	else if (PG_NARGS() == 4)
	{
		/* 文本，文本，文本，布尔 */
		fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
		fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(2));
		fc_fail = PG_GETARG_BOOL(3);
		fc_rconn = fc_getConnectionByName(fc_conname);
	}

	if (!fc_rconn || !fc_rconn->conn)
		fc_dblink_conn_not_avail(fc_conname);

	fc_conn = fc_rconn->conn;

	/* 如果我们不在事务中，启动一个事务 */
	if (PQtransactionStatus(fc_conn) == PQTRANS_IDLE)
	{
		fc_res = PQexec(fc_conn, "BEGIN");
		if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
			fc_dblink_res_internalerror(fc_conn, fc_res, "begin error");
		PQclear(fc_res);
		fc_rconn->newXactForCursor = true;

		/*
		 * 由于事务状态为 IDLE，我们强制游标计数初始为 0。这是必要的，因为
		 * 先前的 ABORT 可能已在不维护游标计数的情况下清除了我们的事务。
		 */
		fc_rconn->openCursorCount = 0;
	}

	/* 如果我们启动了一个事务，增加游标计数 */
	if (fc_rconn->newXactForCursor)
		(fc_rconn->openCursorCount)++;

	appendStringInfo(&fc_buf, "DECLARE %s CURSOR FOR %s", fc_curname, fc_sql);
	fc_res = PQexec(fc_conn, fc_buf.data);
	if (!fc_res || PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		fc_dblink_res_error(fc_conn, fc_conname, fc_res, fc_fail,
						 "while opening cursor \"%s\"", fc_curname);
		PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
	}

	PQclear(fc_res);
	PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

/*
 * 关闭游标
 */
PG_FUNCTION_INFO_V1(dblink_close);
Datum dblink_close(PG_FUNCTION_ARGS)
{
	PGconn	   *fc_conn;
	PGresult   *fc_res = NULL;
	char	   *fc_curname = NULL;
	char	   *fc_conname = NULL;
	StringInfoData fc_buf;
	remoteConn *fc_rconn = NULL;
	bool		fc_fail = true;	/* 默认向后兼容的行为 */

	fc_dblink_init();
	initStringInfo(&fc_buf);

	if (PG_NARGS() == 1)
	{
		/* text */
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_rconn = pconn;
	}
	else if (PG_NARGS() == 2)
	{
		/* 可能是文本，文本或文本，布尔 */
		if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
		{
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_fail = PG_GETARG_BOOL(1);
			fc_rconn = pconn;
		}
		else
		{
			fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
			fc_rconn = fc_getConnectionByName(fc_conname);
		}
	}
	if (PG_NARGS() == 3)
	{
		/* 文本，文本，布尔 */
		fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
		fc_fail = PG_GETARG_BOOL(2);
		fc_rconn = fc_getConnectionByName(fc_conname);
	}

	if (!fc_rconn || !fc_rconn->conn)
		fc_dblink_conn_not_avail(fc_conname);

	fc_conn = fc_rconn->conn;

	appendStringInfo(&fc_buf, "CLOSE %s", fc_curname);

	/* 关闭游标 */
	fc_res = PQexec(fc_conn, fc_buf.data);
	if (!fc_res || PQresultStatus(fc_res) != PGRES_COMMAND_OK)
	{
		fc_dblink_res_error(fc_conn, fc_conname, fc_res, fc_fail,
						 "while closing cursor \"%s\"", fc_curname);
		PG_RETURN_TEXT_P(cstring_to_text("ERROR"));
	}

	PQclear(fc_res);

	/* 如果我们启动了一个事务，减少游标计数 */
	if (fc_rconn->newXactForCursor)
	{
		(fc_rconn->openCursorCount)--;

		/* 如果计数为零，提交事务 */
		if (fc_rconn->openCursorCount == 0)
		{
			fc_rconn->newXactForCursor = false;

			fc_res = PQexec(fc_conn, "COMMIT");
			if (PQresultStatus(fc_res) != PGRES_COMMAND_OK)
				fc_dblink_res_internalerror(fc_conn, fc_res, "commit error");
			PQclear(fc_res);
		}
	}

	PG_RETURN_TEXT_P(cstring_to_text("OK"));
}

/*
 * 从打开的游标中获取结果
 */
PG_FUNCTION_INFO_V1(dblink_fetch);
Datum dblink_fetch(PG_FUNCTION_ARGS)
{
	PGresult   *fc_res = NULL;
	char	   *fc_conname = NULL;
	remoteConn *fc_rconn = NULL;
	PGconn	   *fc_conn = NULL;
	StringInfoData fc_buf;
	char	   *fc_curname = NULL;
	int			fc_howmany = 0;
	bool		fc_fail = true;	/* 默认向后兼容 */

	fc_prepTuplestoreResult(fcinfo);

	fc_dblink_init();

	if (PG_NARGS() == 4)
	{
		/* 文本，文本，整数，布尔 */
		fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
		fc_howmany = PG_GETARG_INT32(2);
		fc_fail = PG_GETARG_BOOL(3);

		fc_rconn = fc_getConnectionByName(fc_conname);
		if (fc_rconn)
			fc_conn = fc_rconn->conn;
	}
	else if (PG_NARGS() == 3)
	{
		/* 文本，文本，整数或文本，整数，布尔 */
		if (get_fn_expr_argtype(fcinfo->flinfo, 2) == BOOLOID)
		{
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_howmany = PG_GETARG_INT32(1);
			fc_fail = PG_GETARG_BOOL(2);
			fc_conn = pconn->conn;
		}
		else
		{
			fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(1));
			fc_howmany = PG_GETARG_INT32(2);

			fc_rconn = fc_getConnectionByName(fc_conname);
			if (fc_rconn)
				fc_conn = fc_rconn->conn;
		}
	}
	else if (PG_NARGS() == 2)
	{
		/* 文本，整数 */
		fc_curname = text_to_cstring(PG_GETARG_TEXT_PP(0));
		fc_howmany = PG_GETARG_INT32(1);
		fc_conn = pconn->conn;
	}

	if (!fc_conn)
		fc_dblink_conn_not_avail(fc_conname);

	initStringInfo(&fc_buf);
	appendStringInfo(&fc_buf, "FETCH %d FROM %s", fc_howmany, fc_curname);

	/*
	 * 尝试执行查询。请注意，由于 libpq 使用 malloc，
	 * PGresult 将长时间存在，即使我们仍在短期限的
	 * 内存上下文中。
	 */
	fc_res = PQexec(fc_conn, fc_buf.data);
	if (!fc_res ||
		(PQresultStatus(fc_res) != PGRES_COMMAND_OK &&
		 PQresultStatus(fc_res) != PGRES_TUPLES_OK))
	{
		fc_dblink_res_error(fc_conn, fc_conname, fc_res, fc_fail,
						 "while fetching from cursor \"%s\"", fc_curname);
		return (Datum) 0;
	}
	else if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
	{
		/* 游标不存在 - 已关闭或名称错误 */
		PQclear(fc_res);
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_CURSOR_NAME),
				 errmsg("cursor \"%s\" does not exist", fc_curname)));
	}

	fc_materializeResult(fcinfo, fc_conn, fc_res);
	return (Datum) 0;
}

/*
 * 注意：这是 dblink 的新推荐版本
 */
PG_FUNCTION_INFO_V1(dblink_record);
Datum dblink_record(PG_FUNCTION_ARGS)
{
	return fc_dblink_record_internal(fcinfo, false);
}

PG_FUNCTION_INFO_V1(dblink_send_query);
Datum dblink_send_query(PG_FUNCTION_ARGS)
{
	PGconn	   *fc_conn;
	char	   *fc_sql;
	int			fc_retval;

	if (PG_NARGS() == 2)
	{
		fc_conn = fc_dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
		fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
	}
	else
		/* 不应该发生 */
		elog(ERROR, "wrong number of arguments");

	/* 异步查询发送 */
	fc_retval = PQsendQuery(fc_conn, fc_sql);
	if (fc_retval != 1)
		elog(NOTICE, "could not send query: %s", pchomp(PQerrorMessage(fc_conn)));

	PG_RETURN_INT32(fc_retval);
}

PG_FUNCTION_INFO_V1(dblink_get_result);
Datum dblink_get_result(PG_FUNCTION_ARGS)
{
	return fc_dblink_record_internal(fcinfo, true);
}

static Datum fc_dblink_record_internal(FunctionCallInfo fcinfo, bool fc_is_async)
{
	PGconn	   *volatile fc_conn = NULL;
	volatile bool fc_freeconn = false;

	fc_prepTuplestoreResult(fcinfo);

	fc_dblink_init();

	PG_TRY();
	{
		char	   *fc_sql = NULL;
		char	   *fc_conname = NULL;
		bool		fc_fail = true;	/* 默认向后兼容 */

		if (!fc_is_async)
		{
			if (PG_NARGS() == 3)
			{
				/* 文本，文本，布尔 */
				fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
				fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
				fc_fail = PG_GETARG_BOOL(2);
				fc_dblink_get_conn(fc_conname, &fc_conn, &fc_conname, &fc_freeconn);
			}
			else if (PG_NARGS() == 2)
			{
				/* 文本,文本或文本,布尔值 */
				if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
				{
					fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
					fc_fail = PG_GETARG_BOOL(1);
					fc_conn = pconn->conn;
				}
				else
				{
					fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
					fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
					fc_dblink_get_conn(fc_conname, &fc_conn, &fc_conname, &fc_freeconn);
				}
			}
			else if (PG_NARGS() == 1)
			{
				/* text */
				fc_conn = pconn->conn;
				fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
			}
			else
				/* 不应该发生 */
				elog(ERROR, "wrong number of arguments");
		}
		else					/* 是否异步 */
		{
			/* 获取异步结果 */
			fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));

			if (PG_NARGS() == 2)
			{
				/* 文本,布尔值 */
				fc_fail = PG_GETARG_BOOL(1);
				fc_conn = fc_dblink_get_named_conn(fc_conname);
			}
			else if (PG_NARGS() == 1)
			{
				/* text */
				fc_conn = fc_dblink_get_named_conn(fc_conname);
			}
			else
				/* 不应该发生 */
				elog(ERROR, "wrong number of arguments");
		}

		if (!fc_conn)
			fc_dblink_conn_not_avail(fc_conname);

		if (!fc_is_async)
		{
			/* 同步查询，使用高效的元组集合方法 */
			fc_materializeQueryResult(fcinfo, fc_conn, fc_conname, fc_sql, fc_fail);
		}
		else
		{
			/* 异步结果获取，按照旧方式进行 */
			PGresult   *fc_res = PQgetResult(fc_conn);

			/* NULL表示我们已经完成所有异步结果 */
			if (fc_res)
			{
				if (PQresultStatus(fc_res) != PGRES_COMMAND_OK &&
					PQresultStatus(fc_res) != PGRES_TUPLES_OK)
				{
					fc_dblink_res_error(fc_conn, fc_conname, fc_res, fc_fail,
									 "while executing query");
					/* 如果未设置fail，将返回空查询结果 */
				}
				else
				{
					fc_materializeResult(fcinfo, fc_conn, fc_res);
				}
			}
		}
	}
	PG_FINALLY();
	{
		/* 如有需要，关闭与数据库的连接 */
		if (fc_freeconn)
		{
			PQfinish(fc_conn);
			ReleaseExternalFD();
		}
	}
	PG_END_TRY();

	return (Datum) 0;
}

/*
 * 验证函数调用者是否可以处理元组存储结果，并为此进行设置。
 *
 * 注意：如果调用者返回而没有实际创建元组存储，执行器将把函数结果视为空集。
 */
static void fc_prepTuplestoreResult(FunctionCallInfo fc_fcinfo)
{
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fc_fcinfo->resultinfo;

	/* 检查查询是否支持我们返回元组存储 */
	if (fc_rsinfo == NULL || !IsA(fc_rsinfo, ReturnSetInfo))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("set-valued function called in context that cannot accept a set")));
	if (!(fc_rsinfo->allowedModes & SFRM_Materialize))
		ereport(ERROR,
				(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
				 errmsg("materialize mode required, but it is not allowed in this context")));

	/* 告诉执行器我们正在返回一个元组存储 */
	fc_rsinfo->returnMode = SFRM_Materialize;

	/* 调用者必须填充这些以返回非空结果 */
	fc_rsinfo->setResult = NULL;
	fc_rsinfo->setDesc = NULL;
}

/*
 * 将PGresult的内容复制到元组存储中，以作为当前函数的结果返回。
 * PGresult将在此函数中释放。
 */
static void fc_materializeResult(FunctionCallInfo fcinfo, PGconn *fc_conn, PGresult *fc_res)
{
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

	/* 必须先调用prepTuplestoreResult */
	Assert(fc_rsinfo->returnMode == SFRM_Materialize);

	PG_TRY();
	{
		TupleDesc	fc_tupdesc;
		bool		fc_is_sql_cmd;
		int			fc_ntuples;
		int			fc_nfields;

		if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
		{
			fc_is_sql_cmd = true;

			/*
			 * 需要一个表示一个TEXT列的元组描述符，以将命令状态字符串
			 * 作为我们的结果元组返回
			 */
			fc_tupdesc = CreateTemplateTupleDesc(1);
			TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "status",
							   TEXTOID, -1, 0);
			fc_ntuples = 1;
			fc_nfields = 1;
		}
		else
		{
			Assert(PQresultStatus(fc_res) == PGRES_TUPLES_OK);

			fc_is_sql_cmd = false;

			/* 获取我们的结果类型的元组描述符 */
			switch (get_call_result_type(fcinfo, NULL, &fc_tupdesc))
			{
				case TYPEFUNC_COMPOSITE:
					/* 成功 */
					break;
				case TYPEFUNC_RECORD:
					/* 未能确定 RECORD 的实际类型 */
					ereport(ERROR,
							(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
							 errmsg("function returning record called in context "
									"that cannot accept type record")));
					break;
				default:
					/* 结果类型不是复合类型 */
					elog(ERROR, "return type must be a row type");
					break;
			}

			/* 确保我们有一个持久的tupdesc副本 */
			fc_tupdesc = CreateTupleDescCopy(fc_tupdesc);
			fc_ntuples = PQntuples(fc_res);
			fc_nfields = PQnfields(fc_res);
		}

		/*
		 * 检查结果和元组描述符具有相同数量的列
		 */
		if (fc_nfields != fc_tupdesc->natts)
			ereport(ERROR,
					(errcode(ERRCODE_DATATYPE_MISMATCH),
					 errmsg("remote query result rowtype does not match "
							"the specified FROM clause rowtype")));

		if (fc_ntuples > 0)
		{
			AttInMetadata *fc_attinmeta;
			int			fc_nestlevel = -1;
			Tuplestorestate *fc_tupstore;
			MemoryContext fc_oldcontext;
			int			fc_row;
			char	  **fc_values;

			fc_attinmeta = TupleDescGetAttInMetadata(fc_tupdesc);

			/* 设置GUC以确保我们正确读取GUC敏感的数据类型 */
			if (!fc_is_sql_cmd)
				fc_nestlevel = fc_applyRemoteGucs(fc_conn);

			fc_oldcontext = MemoryContextSwitchTo(fc_rsinfo->econtext->ecxt_per_query_memory);
			fc_tupstore = tuplestore_begin_heap(true, false, work_mem);
			fc_rsinfo->setResult = fc_tupstore;
			fc_rsinfo->setDesc = fc_tupdesc;
			MemoryContextSwitchTo(fc_oldcontext);

			fc_values = (char **) palloc(fc_nfields * sizeof(char *));

			/* 将所有元组放入元组存储 */
			for (fc_row = 0; fc_row < fc_ntuples; fc_row++)
			{
				HeapTuple	fc_tuple;

				if (!fc_is_sql_cmd)
				{
					int			fc_i;

					for (fc_i = 0; fc_i < fc_nfields; fc_i++)
					{
						if (PQgetisnull(fc_res, fc_row, fc_i))
							fc_values[fc_i] = NULL;
						else
							fc_values[fc_i] = PQgetvalue(fc_res, fc_row, fc_i);
					}
				}
				else
				{
					fc_values[0] = PQcmdStatus(fc_res);
				}

				/* 构建元组并将其放入元组存储中。 */
				fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_values);
				tuplestore_puttuple(fc_tupstore, fc_tuple);
			}

			/* 清理GUC设置，如果我们更改了任何内容 */
			fc_restoreLocalGucs(fc_nestlevel);
		}
	}
	PG_FINALLY();
	{
		/* 确保释放libpq结果 */
		PQclear(fc_res);
	}
	PG_END_TRY();
}

/*
 * 执行给定的SQL命令并将其结果存储到元组存储中，
 * 以作为当前函数的结果返回。
 *
 * 这相当于PQexec后跟materializeResult，但我们利用
 * libpq的单行模式，以避免在转移到元组存储之前在libpq中累积整个结果。
 */
static void fc_materializeQueryResult(FunctionCallInfo fcinfo,
					   PGconn *fc_conn,
					   const char *fc_conname,
					   const char *fc_sql,
					   bool fc_fail)
{
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;
	PGresult   *volatile fc_res = NULL;
	volatile storeInfo fc_sinfo = {0};

	/* 必须先调用prepTuplestoreResult */
	Assert(fc_rsinfo->returnMode == SFRM_Materialize);

	fc_sinfo.fcinfo = fcinfo;

	PG_TRY();
	{
		/* 为数据转换创建短期内存上下文 */
		fc_sinfo.tmpcontext = AllocSetContextCreate(CurrentMemoryContext,
												 "dblink temporary context",
												 ALLOCSET_DEFAULT_SIZES);

		/* 执行查询，将任何元组收集到元组存储中 */
		fc_res = fc_storeQueryResult(&fc_sinfo, fc_conn, fc_sql);

		if (!fc_res ||
			(PQresultStatus(fc_res) != PGRES_COMMAND_OK &&
			 PQresultStatus(fc_res) != PGRES_TUPLES_OK))
		{
			/*
			 * dblink_res_error将清除传递的PGresult，
			 * 所以我们需要这个繁琐的过程，以避免在错误退出期间执行两次
			 */
			PGresult   *fc_res1 = fc_res;

			fc_res = NULL;
			fc_dblink_res_error(fc_conn, fc_conname, fc_res1, fc_fail,
							 "while executing query");
			/* 如果未设置fail，将返回空查询结果 */
		}
		else if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
		{
			/*
			 * storeRow没有被调用，因此我们需要手动将命令
			 * 状态字符串转换为一个元组
			 */
			TupleDesc	fc_tupdesc;
			AttInMetadata *fc_attinmeta;
			Tuplestorestate *fc_tupstore;
			HeapTuple	fc_tuple;
			char	   *fc_values[1];
			MemoryContext fc_oldcontext;

			/*
			 * 需要一个表示一个TEXT列的元组描述符，以将命令状态字符串
			 * 作为我们的结果元组返回
			 */
			fc_tupdesc = CreateTemplateTupleDesc(1);
			TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "status",
							   TEXTOID, -1, 0);
			fc_attinmeta = TupleDescGetAttInMetadata(fc_tupdesc);

			fc_oldcontext = MemoryContextSwitchTo(fc_rsinfo->econtext->ecxt_per_query_memory);
			fc_tupstore = tuplestore_begin_heap(true, false, work_mem);
			fc_rsinfo->setResult = fc_tupstore;
			fc_rsinfo->setDesc = fc_tupdesc;
			MemoryContextSwitchTo(fc_oldcontext);

			fc_values[0] = PQcmdStatus(fc_res);

			/* 构建元组并将其放入元组存储中。 */
			fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_values);
			tuplestore_puttuple(fc_tupstore, fc_tuple);

			PQclear(fc_res);
			fc_res = NULL;
		}
		else
		{
			Assert(PQresultStatus(fc_res) == PGRES_TUPLES_OK);
			/* storeRow 应该已经创建了一个元组存储 */
			Assert(fc_rsinfo->setResult != NULL);

			PQclear(fc_res);
			fc_res = NULL;
		}

		/* 清理数据转换临时内存上下文 */
		if (fc_sinfo.tmpcontext != NULL)
			MemoryContextDelete(fc_sinfo.tmpcontext);
		fc_sinfo.tmpcontext = NULL;

		PQclear(fc_sinfo.last_res);
		fc_sinfo.last_res = NULL;
		PQclear(fc_sinfo.cur_res);
		fc_sinfo.cur_res = NULL;
	}
	PG_CATCH();
	{
		/* 确保释放我们收集的任何 libpq 结果 */
		PQclear(fc_res);
		PQclear(fc_sinfo.last_res);
		PQclear(fc_sinfo.cur_res);
		/* 并清理 libpq 中的任何待处理数据 */
		while ((fc_res = PQgetResult(fc_conn)) != NULL)
			PQclear(fc_res);
		PG_RE_THROW();
	}
	PG_END_TRY();
}

/*
 * 执行查询，并将任何结果行发送到 sinfo->tuplestore。
 */
static PGresult * fc_storeQueryResult(volatile storeInfo *fc_sinfo, PGconn *fc_conn, const char *fc_sql)
{
	bool		fc_first = true;
	int			fc_nestlevel = -1;
	PGresult   *fc_res;

	if (!PQsendQuery(fc_conn, fc_sql))
		elog(ERROR, "could not send query: %s", pchomp(PQerrorMessage(fc_conn)));

	if (!PQsetSingleRowMode(fc_conn))	/* 不应该失败 */
		elog(ERROR, "failed to set single-row mode for dblink query");

	for (;;)
	{
		CHECK_FOR_INTERRUPTS();

		fc_sinfo->cur_res = PQgetResult(fc_conn);
		if (!fc_sinfo->cur_res)
			break;

		if (PQresultStatus(fc_sinfo->cur_res) == PGRES_SINGLE_TUPLE)
		{
			/* 从可能更大的结果集中获得一行 */

			/*
			 * 设置 GUC 以确保我们正确读取 GUC 敏感的数据类型。
			 * 我们不应该在手中没有行之前执行此操作，以确保
			 * libpq 已经看到了之前的 ParameterStatus 协议消息。
			 */
			if (fc_first && fc_nestlevel < 0)
				fc_nestlevel = fc_applyRemoteGucs(fc_conn);

			fc_storeRow(fc_sinfo, fc_sinfo->cur_res, fc_first);

			PQclear(fc_sinfo->cur_res);
			fc_sinfo->cur_res = NULL;
			fc_first = false;
		}
		else
		{
			/* 如果结果集为空，填充元组存储头 */
			if (fc_first && PQresultStatus(fc_sinfo->cur_res) == PGRES_TUPLES_OK)
				fc_storeRow(fc_sinfo, fc_sinfo->cur_res, fc_first);

			/* 将完成的结果存储在 last_res 中 */
			PQclear(fc_sinfo->last_res);
			fc_sinfo->last_res = fc_sinfo->cur_res;
			fc_sinfo->cur_res = NULL;
			fc_first = true;
		}
	}

	/* 清理GUC设置，如果我们更改了任何内容 */
	fc_restoreLocalGucs(fc_nestlevel);

	/* 返回 last_res */
	fc_res = fc_sinfo->last_res;
	fc_sinfo->last_res = NULL;
	return fc_res;
}

/*
 * 将单行发送到 sinfo->tuplestore。
 *
 * 如果 "first" 为真，则使用 PGresult 的元数据创建元组存储
 * （在这种情况下，PGresult 可能包含零行或一行）。
 */
static void fc_storeRow(volatile storeInfo *fc_sinfo, PGresult *fc_res, bool fc_first)
{
	int			fc_nfields = PQnfields(fc_res);
	HeapTuple	fc_tuple;
	int			fc_i;
	MemoryContext fc_oldcontext;

	if (fc_first)
	{
		/* 为新的结果集做好准备 */
		ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fc_sinfo->fcinfo->resultinfo;
		TupleDesc	fc_tupdesc;

		/*
		 * 如果查询字符串包含多个 SQL 命令，则可能会获取多个结果集。
		 * 在这种情况下，我们遵循 PQexec 的传统行为，丢弃除最后一个结果以外的所有结果。
		 */
		if (fc_sinfo->tuplestore)
			tuplestore_end(fc_sinfo->tuplestore);
		fc_sinfo->tuplestore = NULL;

		/* 获取我们的结果类型的元组描述符 */
		switch (get_call_result_type(fc_sinfo->fcinfo, NULL, &fc_tupdesc))
		{
			case TYPEFUNC_COMPOSITE:
				/* 成功 */
				break;
			case TYPEFUNC_RECORD:
				/* 未能确定 RECORD 的实际类型 */
				ereport(ERROR,
						(errcode(ERRCODE_FEATURE_NOT_SUPPORTED),
						 errmsg("function returning record called in context "
								"that cannot accept type record")));
				break;
			default:
				/* 结果类型不是复合类型 */
				elog(ERROR, "return type must be a row type");
				break;
		}

		/* 确保我们有一个持久的tupdesc副本 */
		fc_tupdesc = CreateTupleDescCopy(fc_tupdesc);

		/* 检查结果和元组描述符具有相同数量的列 */
		if (fc_nfields != fc_tupdesc->natts)
			ereport(ERROR,
					(errcode(ERRCODE_DATATYPE_MISMATCH),
					 errmsg("remote query result rowtype does not match "
							"the specified FROM clause rowtype")));

		/* 准备 attinmeta 以便后续数据转换 */
		fc_sinfo->attinmeta = TupleDescGetAttInMetadata(fc_tupdesc);

		/* 创建一个新的、空的元组存储 */
		fc_oldcontext = MemoryContextSwitchTo(fc_rsinfo->econtext->ecxt_per_query_memory);
		fc_sinfo->tuplestore = tuplestore_begin_heap(true, false, work_mem);
		fc_rsinfo->setResult = fc_sinfo->tuplestore;
		fc_rsinfo->setDesc = fc_tupdesc;
		MemoryContextSwitchTo(fc_oldcontext);

		/* 如果结果集为空则完成 */
		if (PQntuples(fc_res) == 0)
			return;

		/*
		 * 设置足够宽的字符串指针数组；这个数组的大小不会改变，所以很容易预分配。
		 */
		if (fc_sinfo->cstrs)
			pfree(fc_sinfo->cstrs);
		fc_sinfo->cstrs = (char **) palloc(fc_nfields * sizeof(char *));
	}

	/* 如果我们到这里，应该有一个单行结果 */
	Assert(PQntuples(fc_res) == 1);

	/*
	 * 在一个临时上下文中执行以下工作，我们在每个元组之后重置它。
	 * 这不仅清理我们可以直接访问的数据，还清理 I/O 函数可能泄漏的任何杂质。
	 */
	fc_oldcontext = MemoryContextSwitchTo(fc_sinfo->tmpcontext);

	/*
	 * 用以 null 结尾的字符串填充 cstrs，其值为列值。
	 */
	for (fc_i = 0; fc_i < fc_nfields; fc_i++)
	{
		if (PQgetisnull(fc_res, 0, fc_i))
			fc_sinfo->cstrs[fc_i] = NULL;
		else
			fc_sinfo->cstrs[fc_i] = PQgetvalue(fc_res, 0, fc_i);
	}

	/* 将行转换为元组，并将其添加到元组存储中 */
	fc_tuple = BuildTupleFromCStrings(fc_sinfo->attinmeta, fc_sinfo->cstrs);

	tuplestore_puttuple(fc_sinfo->tuplestore, fc_tuple);

	/* 清理 */
	MemoryContextSwitchTo(fc_oldcontext);
	MemoryContextReset(fc_sinfo->tmpcontext);
}

/*
 * 按名称列出所有打开的 dblink 连接。
 * 返回所有连接名称的数组。
 * 不接受参数
 */
PG_FUNCTION_INFO_V1(dblink_get_connections);
Datum dblink_get_connections(PG_FUNCTION_ARGS)
{
	HASH_SEQ_STATUS fc_status;
	remoteConnHashEnt *fc_hentry;
	ArrayBuildState *fc_astate = NULL;

	if (remoteConnHash)
	{
		hash_seq_init(&fc_status, remoteConnHash);
		while ((fc_hentry = (remoteConnHashEnt *) hash_seq_search(&fc_status)) != NULL)
		{
			/* 储存当前值 */
			fc_astate = accumArrayResult(fc_astate,
									  CStringGetTextDatum(fc_hentry->name),
									  false, TEXTOID, CurrentMemoryContext);
		}
	}

	if (fc_astate)
		PG_RETURN_ARRAYTYPE_P(makeArrayResult(fc_astate,
											  CurrentMemoryContext));
	else
		PG_RETURN_NULL();
}

/*
 * 检查给定的远程连接是否繁忙
 *
 * 如果连接繁忙则返回 1，否则返回 0
 * 参数：
 *	text connection_name - 要检查的连接名称
 *
 */
PG_FUNCTION_INFO_V1(dblink_is_busy);
Datum dblink_is_busy(PG_FUNCTION_ARGS)
{
	PGconn	   *fc_conn;

	fc_dblink_init();
	fc_conn = fc_dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));

	PQconsumeInput(fc_conn);
	PG_RETURN_INT32(PQisBusy(fc_conn));
}

/*
 * 取消连接上正在进行的请求
 *
 * 返回文本：
 *	"OK" 如果取消请求已正确发送，
 *		否则返回错误消息
 *
 * 参数：
 *	text connection_name - 要检查的连接名称
 *
 */
PG_FUNCTION_INFO_V1(dblink_cancel_query);
Datum dblink_cancel_query(PG_FUNCTION_ARGS)
{
	int			fc_res;
	PGconn	   *fc_conn;
	PGcancel   *fc_cancel;
	char		fc_errbuf[256];

	fc_dblink_init();
	fc_conn =fc_dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
	fc_cancel = PQgetCancel(fc_conn);

	fc_res = PQcancel(fc_cancel, fc_errbuf, 256);
	PQfreeCancel(fc_cancel);

	if (fc_res == 1)
		PG_RETURN_TEXT_P(cstring_to_text("OK"));
	else
		PG_RETURN_TEXT_P(cstring_to_text(fc_errbuf));
}


/*
 * 从连接获取错误信息
 *
 * 返回文本：
 *	"OK" 如果没有错误，否则返回错误信息
 *
 * 参数：
 *	text connection_name - 要检查的连接名称
 *
 */
PG_FUNCTION_INFO_V1(dblink_error_message);
Datum dblink_error_message(PG_FUNCTION_ARGS)
{
	char	   *fc_msg;
	PGconn	   *fc_conn;

	fc_dblink_init();
	fc_conn = fc_dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));

	fc_msg = PQerrorMessage(fc_conn);
	if (fc_msg == NULL || fc_msg[0] == '\0')
		PG_RETURN_TEXT_P(cstring_to_text("OK"));
	else
		PG_RETURN_TEXT_P(cstring_to_text(pchomp(fc_msg)));
}

/*
 * 执行一个 SQL 非 SELECT 命令
 */
PG_FUNCTION_INFO_V1(dblink_exec);
Datum dblink_exec(PG_FUNCTION_ARGS)
{
	text	   *volatile fc_sql_cmd_status = NULL;
	PGconn	   *volatile fc_conn = NULL;
	volatile bool fc_freeconn = false;

	fc_dblink_init();

	PG_TRY();
	{
		PGresult   *fc_res = NULL;
		char	   *fc_sql = NULL;
		char	   *fc_conname = NULL;
		bool		fc_fail = true;	/* 默认向后兼容的行为 */

		if (PG_NARGS() == 3)
		{
			/* 必须是 text,text,bool */
			fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
			fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
			fc_fail = PG_GETARG_BOOL(2);
			fc_dblink_get_conn(fc_conname, &fc_conn, &fc_conname, &fc_freeconn);
		}
		else if (PG_NARGS() == 2)
		{
			/* 可能是文本，文本或文本，布尔 */
			if (get_fn_expr_argtype(fcinfo->flinfo, 1) == BOOLOID)
			{
				fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
				fc_fail = PG_GETARG_BOOL(1);
				fc_conn = pconn->conn;
			}
			else
			{
				fc_conname = text_to_cstring(PG_GETARG_TEXT_PP(0));
				fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(1));
				fc_dblink_get_conn(fc_conname, &fc_conn, &fc_conname, &fc_freeconn);
			}
		}
		else if (PG_NARGS() == 1)
		{
			/* 必须是单个文本参数 */
			fc_conn = pconn->conn;
			fc_sql = text_to_cstring(PG_GETARG_TEXT_PP(0));
		}
		else
			/* 不应该发生 */
			elog(ERROR, "wrong number of arguments");

		if (!fc_conn)
			fc_dblink_conn_not_avail(fc_conname);

		fc_res = PQexec(fc_conn, fc_sql);
		if (!fc_res ||
			(PQresultStatus(fc_res) != PGRES_COMMAND_OK &&
			 PQresultStatus(fc_res) != PGRES_TUPLES_OK))
		{
			fc_dblink_res_error(fc_conn, fc_conname, fc_res, fc_fail,
							 "while executing command");

			/*
			 * 并保存命令状态字符串的副本以返回作为我们的
			 * 结果元组
			 */
			fc_sql_cmd_status = cstring_to_text("ERROR");
		}
		else if (PQresultStatus(fc_res) == PGRES_COMMAND_OK)
		{
			/*
			 * 并保存命令状态字符串的副本以返回作为我们的
			 * 结果元组
			 */
			fc_sql_cmd_status = cstring_to_text(PQcmdStatus(fc_res));
			PQclear(fc_res);
		}
		else
		{
			PQclear(fc_res);
			ereport(ERROR,
					(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
					 errmsg("statement returning results not allowed")));
		}
	}
	PG_FINALLY();
	{
		/* 如有需要，关闭与数据库的连接 */
		if (fc_freeconn)
		{
			PQfinish(fc_conn);
			ReleaseExternalFD();
		}
	}
	PG_END_TRY();

	PG_RETURN_TEXT_P(fc_sql_cmd_status);
}


/*
 * dblink_get_pkey
 *
 * 返回提供关系的主键字段列表，
 * 如果不存在则返回 NULL。
 */
PG_FUNCTION_INFO_V1(dblink_get_pkey);
Datum dblink_get_pkey(PG_FUNCTION_ARGS)
{
	int16		fc_indnkeyatts;
	char	  **fc_results;
	FuncCallContext *fc_funcctx;
	int32		fc_call_cntr;
	int32		fc_max_calls;
	AttInMetadata *fc_attinmeta;
	MemoryContext fc_oldcontext;

	/* 仅在函数的第一次调用时完成的工作 */
	if (SRF_IS_FIRSTCALL())
	{
		Relation	fc_rel;
		TupleDesc	fc_tupdesc;

		/* 为跨调用持久性创建一个函数上下文 */
		fc_funcctx = SRF_FIRSTCALL_INIT();

		/*
		 * 切换到适合多次函数调用的内存上下文
		 */
		fc_oldcontext = MemoryContextSwitchTo(fc_funcctx->multi_call_memory_ctx);

		/* 打开目标关系 */
		fc_rel = fc_get_rel_from_relname(PG_GETARG_TEXT_PP(0), AccessShareLock, ACL_SELECT);

		/* 获取 attnums 数组 */
		fc_results = fc_get_pkey_attnames(fc_rel, &fc_indnkeyatts);

		relation_close(fc_rel, AccessShareLock);

		/*
		 * 需要一个表示一个 INT 和一个 TEXT 列的元组描述符
		 */
		fc_tupdesc = CreateTemplateTupleDesc(2);
		TupleDescInitEntry(fc_tupdesc, (AttrNumber) 1, "position",
						   INT4OID, -1, 0);
		TupleDescInitEntry(fc_tupdesc, (AttrNumber) 2, "colname",
						   TEXTOID, -1, 0);

		/*
		 * 生成稍后从原始 C 字符串生成元组所需的属性元数据
		 */
		fc_attinmeta = TupleDescGetAttInMetadata(fc_tupdesc);
		fc_funcctx->attinmeta = fc_attinmeta;

		if ((fc_results != NULL) && (fc_indnkeyatts > 0))
		{
			fc_funcctx->max_calls = fc_indnkeyatts;

			/* 得到结果，跟踪它们 */
			fc_funcctx->user_fctx = fc_results;
		}
		else
		{
			/* 当没有结果时的快速处理 */
			MemoryContextSwitchTo(fc_oldcontext);
			SRF_RETURN_DONE(fc_funcctx);
		}

		MemoryContextSwitchTo(fc_oldcontext);
	}

	/* 每次调用函数时完成的工作 */
	fc_funcctx = SRF_PERCALL_SETUP();

	/*
	 * 初始化每次调用的变量
	 */
	fc_call_cntr = fc_funcctx->call_cntr;
	fc_max_calls = fc_funcctx->max_calls;

	fc_results = (char **) fc_funcctx->user_fctx;
	fc_attinmeta = fc_funcctx->attinmeta;

	if (fc_call_cntr < fc_max_calls)	/* 当还有更多内容要发送时执行 */
	{
		char	  **fc_values;
		HeapTuple	fc_tuple;
		Datum		fc_result;

		fc_values = (char **) palloc(2 * sizeof(char *));
		fc_values[0] = psprintf("%d", fc_call_cntr + 1);
		fc_values[1] = fc_results[fc_call_cntr];

		/* 构建元组 */
		fc_tuple = BuildTupleFromCStrings(fc_attinmeta, fc_values);

		/* 将元组转化为一个 datum */
		fc_result = HeapTupleGetDatum(fc_tuple);

		SRF_RETURN_NEXT(fc_funcctx, fc_result);
	}
	else
	{
		/* 当没有更多内容时执行 */
		SRF_RETURN_DONE(fc_funcctx);
	}
}


/*
 * dblink_build_sql_insert
 *
 * 用于生成基于本地关系中现有元组的 SQL 插入语句。
 * 这对于通过 dblink 有选择地复制数据到另一个服务器非常有用。
 *
 * API：
 * <relname> - 相关本地表的名称
 * <pkattnums> - 一个 int2vector 的 attnums，将用于
 * 以识别相关的本地元组
 * <pknumatts> - pkattnums 中的 attnums 数量
 * <src_pkattvals_arry> - 键值的文本数组，将用于
 * 识别相关的本地元组
 * <tgt_pkattvals_arry> - 键值的文本数组，将用于
 * 构建远程执行的字符串。这些将被替换为
 * src_pkattvals_arry 中的对应值
 */
PG_FUNCTION_INFO_V1(dblink_build_sql_insert);
Datum dblink_build_sql_insert(PG_FUNCTION_ARGS)
{
	text	   *fc_relname_text = PG_GETARG_TEXT_PP(0);
	int2vector *fc_pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
	int32		fc_pknumatts_arg = PG_GETARG_INT32(2);
	ArrayType  *fc_src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
	ArrayType  *fc_tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
	Relation	fc_rel;
	int		   *fc_pkattnums;
	int			fc_pknumatts;
	char	  **fc_src_pkattvals;
	char	  **fc_tgt_pkattvals;
	int			fc_src_nitems;
	int			fc_tgt_nitems;
	char	   *fc_sql;

	/*
	 * 打开目标关系。
	 */
	fc_rel = fc_get_rel_from_relname(fc_relname_text, AccessShareLock, ACL_SELECT);

	/*
	 * 处理 pkattnums 参数。
	 */
	fc_validate_pkattnums(fc_rel, fc_pkattnums_arg, fc_pknumatts_arg,
					   &fc_pkattnums, &fc_pknumatts);

	/*
	 * 源数组由将用于从本地系统定位
	 * 相关元组的键值组成。
	 */
	fc_src_pkattvals = fc_get_text_array_contents(fc_src_pkattvals_arry, &fc_src_nitems);

	/*
	 * 每个键 attnum 应该有一个源数组键值
	 */
	if (fc_src_nitems != fc_pknumatts)
		ereport(ERROR,
				(errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
				 errmsg("source key array length must match number of key attributes")));

	/*
	 * 目标数组由将用于构建
	 * 用于远程系统的 SQL 字符串的键值组成。
	 */
	fc_tgt_pkattvals = fc_get_text_array_contents(fc_tgt_pkattvals_arry, &fc_tgt_nitems);

	/*
	 * 每个键 attnum 应该有一个目标数组键值
	 */
	if (fc_tgt_nitems != fc_pknumatts)
		ereport(ERROR,
				(errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
				 errmsg("target key array length must match number of key attributes")));

	/*
	 * 准备工作终于完成。去获取 SQL 字符串。
	 */
	fc_sql = fc_get_sql_insert(fc_rel, fc_pkattnums, fc_pknumatts, fc_src_pkattvals, fc_tgt_pkattvals);

	/*
	 * 现在我们可以关闭关系。
	 */
	relation_close(fc_rel, AccessShareLock);

	/*
	 * 并发送它
	 */
	PG_RETURN_TEXT_P(cstring_to_text(fc_sql));
}


/*
 * dblink_build_sql_delete
 *
 * 用于生成一个 SQL 删除语句。
 * 这对于通过 dblink 有选择地将删除复制到另一个服务器非常有用。
 *
 * API:
 * <relname> - 远程表的名称
 * <pkattnums> - 一个 int2vector 的 attnums，将用于
 * 识别所关心的远程元组
 * <pknumatts> - pkattnums 中 attnums 的数量
 * <tgt_pkattvals_arry> - 将用于构建字符串以远程执行的键值文本数组。
 */
PG_FUNCTION_INFO_V1(dblink_build_sql_delete);
Datum dblink_build_sql_delete(PG_FUNCTION_ARGS)
{
	text	   *fc_relname_text = PG_GETARG_TEXT_PP(0);
	int2vector *fc_pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
	int32		fc_pknumatts_arg = PG_GETARG_INT32(2);
	ArrayType  *fc_tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
	Relation	fc_rel;
	int		   *fc_pkattnums;
	int			fc_pknumatts;
	char	  **fc_tgt_pkattvals;
	int			fc_tgt_nitems;
	char	   *fc_sql;

	/*
	 * 打开目标关系。
	 */
	fc_rel = fc_get_rel_from_relname(fc_relname_text, AccessShareLock, ACL_SELECT);

	/*
	 * 处理 pkattnums 参数。
	 */
	fc_validate_pkattnums(fc_rel, fc_pkattnums_arg, fc_pknumatts_arg,
					   &fc_pkattnums, &fc_pknumatts);

	/*
	 * 目标数组由将用于构建
	 * 用于远程系统的 SQL 字符串的键值组成。
	 */
	fc_tgt_pkattvals = fc_get_text_array_contents(fc_tgt_pkattvals_arry, &fc_tgt_nitems);

	/*
	 * 每个键 attnum 应该有一个目标数组键值
	 */
	if (fc_tgt_nitems != fc_pknumatts)
		ereport(ERROR,
				(errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
				 errmsg("target key array length must match number of key attributes")));

	/*
	 * 准备工作终于完成。去获取 SQL 字符串。
	 */
	fc_sql = fc_get_sql_delete(fc_rel, fc_pkattnums, fc_pknumatts, fc_tgt_pkattvals);

	/*
	 * 现在我们可以关闭关系。
	 */
	relation_close(fc_rel, AccessShareLock);

	/*
	 * 并发送它
	 */
	PG_RETURN_TEXT_P(cstring_to_text(fc_sql));
}


/*
 * dblink_build_sql_update
 *
 * 用于根据本地关系中的现有元组生成 SQL 更新语句。
 * 这对于通过 dblink 有选择地将数据复制到另一个服务器非常有用。
 *
 * API:
 * <relname> - 本地表的名称
 * <pkattnums> - 一个 int2vector 的 attnums，将用于
 * 识别所关心的本地元组
 * <pknumatts> - pkattnums 中 attnums 的数量
 * <src_pkattvals_arry> - 将用于识别所关心的本地元组的键值文本数组
 * <tgt_pkattvals_arry> - 将用于构建字符串以远程执行的键值文本数组。这些被替换
 * 为 src_pkattvals_arry 中的对应值
 */
PG_FUNCTION_INFO_V1(dblink_build_sql_update);
Datum dblink_build_sql_update(PG_FUNCTION_ARGS)
{
	text	   *fc_relname_text = PG_GETARG_TEXT_PP(0);
	int2vector *fc_pkattnums_arg = (int2vector *) PG_GETARG_POINTER(1);
	int32		fc_pknumatts_arg = PG_GETARG_INT32(2);
	ArrayType  *fc_src_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(3);
	ArrayType  *fc_tgt_pkattvals_arry = PG_GETARG_ARRAYTYPE_P(4);
	Relation	fc_rel;
	int		   *fc_pkattnums;
	int			fc_pknumatts;
	char	  **fc_src_pkattvals;
	char	  **fc_tgt_pkattvals;
	int			fc_src_nitems;
	int			fc_tgt_nitems;
	char	   *fc_sql;

	/*
	 * 打开目标关系。
	 */
	fc_rel = fc_get_rel_from_relname(fc_relname_text, AccessShareLock, ACL_SELECT);

	/*
	 * 处理 pkattnums 参数。
	 */
	fc_validate_pkattnums(fc_rel, fc_pkattnums_arg, fc_pknumatts_arg,
					   &fc_pkattnums, &fc_pknumatts);

	/*
	 * 源数组由将用于从本地系统定位
	 * 相关元组的键值组成。
	 */
	fc_src_pkattvals = fc_get_text_array_contents(fc_src_pkattvals_arry, &fc_src_nitems);

	/*
	 * 每个键 attnum 应该有一个源数组键值
	 */
	if (fc_src_nitems != fc_pknumatts)
		ereport(ERROR,
				(errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
				 errmsg("source key array length must match number of key attributes")));

	/*
	 * 目标数组由将用于构建
	 * 用于远程系统的 SQL 字符串的键值组成。
	 */
	fc_tgt_pkattvals = fc_get_text_array_contents(fc_tgt_pkattvals_arry, &fc_tgt_nitems);

	/*
	 * 每个键 attnum 应该有一个目标数组键值
	 */
	if (fc_tgt_nitems != fc_pknumatts)
		ereport(ERROR,
				(errcode(ERRCODE_ARRAY_SUBSCRIPT_ERROR),
				 errmsg("target key array length must match number of key attributes")));

	/*
	 * 准备工作终于完成。去获取 SQL 字符串。
	 */
	fc_sql = fc_get_sql_update(fc_rel, fc_pkattnums, fc_pknumatts, fc_src_pkattvals, fc_tgt_pkattvals);

	/*
	 * 现在我们可以关闭关系。
	 */
	relation_close(fc_rel, AccessShareLock);

	/*
	 * 并发送它
	 */
	PG_RETURN_TEXT_P(cstring_to_text(fc_sql));
}

/*
 * dblink_current_query
 * 返回当前查询字符串
 * 以便在（其他事项包括）重写规则中使用
 */
PG_FUNCTION_INFO_V1(dblink_current_query);
Datum dblink_current_query(PG_FUNCTION_ARGS)
{
	/* 这现在仅仅是内置函数 current_query() 的别名 */
	PG_RETURN_DATUM(current_query(fcinfo));
}

/*
 * 检索连接的异步通知。
 *
 * 返回通知的记录集，或如果没有接收到则返回空集。
 * 可以选择性地将命名连接作为参数，但默认使用未命名的
 * 连接。
 *
 */
#define DBLINK_NOTIFY_COLS		3

PG_FUNCTION_INFO_V1(dblink_get_notify);
Datum dblink_get_notify(PG_FUNCTION_ARGS)
{
	PGconn	   *fc_conn;
	PGnotify   *fc_notify;
	ReturnSetInfo *fc_rsinfo = (ReturnSetInfo *) fcinfo->resultinfo;

	fc_dblink_init();
	if (PG_NARGS() == 1)
		fc_conn = fc_dblink_get_named_conn(text_to_cstring(PG_GETARG_TEXT_PP(0)));
	else
		fc_conn = pconn->conn;

	InitMaterializedSRF(fcinfo, 0);

	PQconsumeInput(fc_conn);
	while ((fc_notify = PQnotifies(fc_conn)) != NULL)
	{
		Datum		fc_values[DBLINK_NOTIFY_COLS];
		bool		fc_nulls[DBLINK_NOTIFY_COLS];

		memset(fc_values, 0, sizeof(fc_values));
		memset(fc_nulls, 0, sizeof(fc_nulls));

		if (fc_notify->relname != NULL)
			fc_values[0] = CStringGetTextDatum(fc_notify->relname);
		else
			fc_nulls[0] = true;

		fc_values[1] = Int32GetDatum(fc_notify->be_pid);

		if (fc_notify->extra != NULL)
			fc_values[2] = CStringGetTextDatum(fc_notify->extra);
		else
			fc_nulls[2] = true;

		tuplestore_putvalues(fc_rsinfo->setResult, fc_rsinfo->setDesc, fc_values, fc_nulls);

		PQfreemem(fc_notify);
		PQconsumeInput(fc_conn);
	}

	return (Datum) 0;
}

/*
 * 验证给定给 dblink 外部服务器或用户映射的选项。
 * 如果任何选项无效，则引发错误。
 *
 * 我们只检查这里的选项名称，因此选项中的语义错误，
 * 如无效的数字格式，将在尝试连接时被检测到。
 */
PG_FUNCTION_INFO_V1(dblink_fdw_validator);
Datum dblink_fdw_validator(PG_FUNCTION_ARGS)
{
	List	   *fc_options_list = untransformRelOptions(PG_GETARG_DATUM(0));
	Oid			fc_context = PG_GETARG_OID(1);
	ListCell   *fc_cell;

	static const PQconninfoOption *fc_options = NULL;

	/*
	 * 获取有效的 libpq 选项列表。
	 *
	 * 为了避免不必要的工作，我们只获取一次列表，并在
	 * 本后端进程的生命周期内使用它。我们无需担心
	 * 内存上下文问题，因为 PQconndefaults 使用 malloc 分配。
	 */
	if (!fc_options)
	{
		fc_options = PQconndefaults();
		if (!fc_options)			/* 假设失败的原因是 OOM */
			ereport(ERROR,
					(errcode(ERRCODE_FDW_OUT_OF_MEMORY),
					 errmsg("out of memory"),
					 errdetail("Could not get libpq's default connection options.")));
	}

	/* 验证每个提供的选项。 */
	foreach(fc_cell, fc_options_list)
	{
		DefElem    *fc_def = (DefElem *) lfirst(fc_cell);

		if (!fc_is_valid_dblink_option(fc_options, fc_def->defname, fc_context))
		{
			/*
			 * 未知选项，或针对指定上下文的无效选项，因此
			 * 抱怨它。提供一个有效选项的提示列表
			 * 以供该上下文参考。
			 */
			StringInfoData fc_buf;
			const PQconninfoOption *fc_opt;

			initStringInfo(&fc_buf);
			for (fc_opt = fc_options; fc_opt->keyword; fc_opt++)
			{
				if (fc_is_valid_dblink_option(fc_options, fc_opt->keyword, fc_context))
					appendStringInfo(&fc_buf, "%s%s",
									 (fc_buf.len > 0) ? ", " : "",
									 fc_opt->keyword);
			}
			ereport(ERROR,
					(errcode(ERRCODE_FDW_OPTION_NAME_NOT_FOUND),
					 errmsg("invalid option \"%s\"", fc_def->defname),
					 fc_buf.len > 0
					 ? errhint("Valid options in this context are: %s",
							   fc_buf.data)
					 : errhint("There are no valid options in this context.")));
		}
	}

	PG_RETURN_VOID();
}


/*************************************************************
 * 内部函数
 */


/*
 * get_pkey_attnames
 *
 * 获取给定关系的主键 attnames。
 * 如果没有主键，则返回 NULL，并将 indnkeyatts 设置为 0。
 */
static char ** fc_get_pkey_attnames(Relation fc_rel, int16 *fc_indnkeyatts)
{
	Relation	fc_indexRelation;
	ScanKeyData fc_skey;
	SysScanDesc fc_scan;
	HeapTuple	fc_indexTuple;
	int			fc_i;
	char	  **fc_result = NULL;
	TupleDesc	fc_tupdesc;

	/* 如果没有主键存在，初始化 indnkeyatts 为 0 */
	*fc_indnkeyatts = 0;

	fc_tupdesc = fc_rel->rd_att;

	/* 准备扫描 pg_index 以查找 indrelid = 该关系的条目。 */
	fc_indexRelation = table_open(IndexRelationId, AccessShareLock);
	ScanKeyInit(&fc_skey,
				Anum_pg_index_indrelid,
				BTEqualStrategyNumber, F_OIDEQ,
				ObjectIdGetDatum(RelationGetRelid(fc_rel)));

	fc_scan = systable_beginscan(fc_indexRelation, IndexIndrelidIndexId, true,
							  NULL, 1, &fc_skey);

	while (HeapTupleIsValid(fc_indexTuple = systable_getnext(fc_scan)))
	{
		Form_pg_index fc_index = (Form_pg_index) GETSTRUCT(fc_indexTuple);

		/* 我们只对主键感兴趣 */
		if (fc_index->indisprimary)
		{
			*fc_indnkeyatts = fc_index->indnkeyatts;
			if (*fc_indnkeyatts > 0)
			{
				fc_result = (char **) palloc(*fc_indnkeyatts * sizeof(char *));

				for (fc_i = 0; fc_i < *fc_indnkeyatts; fc_i++)
					fc_result[fc_i] = SPI_fname(fc_tupdesc, fc_index->indkey.values[fc_i]);
			}
			break;
		}
	}

	systable_endscan(fc_scan);
	table_close(fc_indexRelation, AccessShareLock);

	return fc_result;
}

/*
 * 将 text[] 反构造成 C 字符串（请注意，任何 NULL 元素将
 * 作为 NULL 指针返回）
 */
static char ** fc_get_text_array_contents(ArrayType *fc_array, int *fc_numitems)
{
	int			fc_ndim = ARR_NDIM(fc_array);
	int		   *fc_dims = ARR_DIMS(fc_array);
	int			fc_nitems;
	int16		fc_typlen;
	bool		fc_typbyval;
	char		fc_typalign;
	char	  **fc_values;
	char	   *fc_ptr;
	bits8	   *fc_bitmap;
	int			fc_bitmask;
	int			fc_i;

	Assert(ARR_ELEMTYPE(fc_array) == TEXTOID);

	*fc_numitems = fc_nitems = ArrayGetNItems(fc_ndim, fc_dims);

	get_typlenbyvalalign(ARR_ELEMTYPE(fc_array),
						 &fc_typlen, &fc_typbyval, &fc_typalign);

	fc_values = (char **) palloc(fc_nitems * sizeof(char *));

	fc_ptr = ARR_DATA_PTR(fc_array);
	fc_bitmap = ARR_NULLBITMAP(fc_array);
	fc_bitmask = 1;

	for (fc_i = 0; fc_i < fc_nitems; fc_i++)
	{
		if (fc_bitmap && (*fc_bitmap & fc_bitmask) == 0)
		{
			fc_values[fc_i] = NULL;
		}
		else
		{
			fc_values[fc_i] = TextDatumGetCString(PointerGetDatum(fc_ptr));
			fc_ptr = att_addlength_pointer(fc_ptr, fc_typlen, fc_ptr);
			fc_ptr = (char *) att_align_nominal(fc_ptr, fc_typalign);
		}

		/* 如果有，向前推进位图指针 */
		if (fc_bitmap)
		{
			fc_bitmask <<= 1;
			if (fc_bitmask == 0x100)
			{
				fc_bitmap++;
				fc_bitmask = 1;
			}
		}
	}

	return fc_values;
}

static char * fc_get_sql_insert(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals, char **fc_tgt_pkattvals)
{
	char	   *fc_relname;
	HeapTuple	fc_tuple;
	TupleDesc	fc_tupdesc;
	int			fc_natts;
	StringInfoData fc_buf;
	char	   *fc_val;
	int			fc_key;
	int			fc_i;
	bool		fc_needComma;

	initStringInfo(&fc_buf);

	/* 获取关系名称，包括所需的模式前缀和引用 */
	fc_relname = fc_generate_relation_name(fc_rel);

	fc_tupdesc = fc_rel->rd_att;
	fc_natts = fc_tupdesc->natts;

	fc_tuple = fc_get_tuple_of_interest(fc_rel, fc_pkattnums, fc_pknumatts, fc_src_pkattvals);
	if (!fc_tuple)
		ereport(ERROR,
				(errcode(ERRCODE_CARDINALITY_VIOLATION),
				 errmsg("source row not found")));

	appendStringInfo(&fc_buf, "INSERT INTO %s(", fc_relname);

	fc_needComma = false;
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		Form_pg_attribute fc_att = TupleDescAttr(fc_tupdesc, fc_i);

		if (fc_att->attisdropped)
			continue;

		if (fc_needComma)
			appendStringInfoChar(&fc_buf, ',');

		appendStringInfoString(&fc_buf,
							   fc_quote_ident_cstr(NameStr(fc_att->attname)));
		fc_needComma = true;
	}

	appendStringInfoString(&fc_buf, ") VALUES(");

	/*
	 * 注意：i 是物理列号（从 0 开始计数）。
	 */
	fc_needComma = false;
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		if (TupleDescAttr(fc_tupdesc, fc_i)->attisdropped)
			continue;

		if (fc_needComma)
			appendStringInfoChar(&fc_buf, ',');

		fc_key = fc_get_attnum_pk_pos(fc_pkattnums, fc_pknumatts, fc_i);

		if (fc_key >= 0)
			fc_val = fc_tgt_pkattvals[fc_key] ? pstrdup(fc_tgt_pkattvals[fc_key]) : NULL;
		else
			fc_val = SPI_getvalue(fc_tuple, fc_tupdesc, fc_i + 1);

		if (fc_val != NULL)
		{
			appendStringInfoString(&fc_buf, quote_literal_cstr(fc_val));
			pfree(fc_val);
		}
		else
			appendStringInfoString(&fc_buf, "NULL");
		fc_needComma = true;
	}
	appendStringInfoChar(&fc_buf, ')');

	return fc_buf.data;
}

static char * fc_get_sql_delete(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_tgt_pkattvals)
{
	char	   *fc_relname;
	TupleDesc	fc_tupdesc;
	StringInfoData fc_buf;
	int			fc_i;

	initStringInfo(&fc_buf);

	/* 获取关系名称，包括所需的模式前缀和引用 */
	fc_relname = fc_generate_relation_name(fc_rel);

	fc_tupdesc = fc_rel->rd_att;

	appendStringInfo(&fc_buf, "DELETE FROM %s WHERE ", fc_relname);
	for (fc_i = 0; fc_i < fc_pknumatts; fc_i++)
	{
		int			fc_pkattnum = fc_pkattnums[fc_i];
		Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_pkattnum);

		if (fc_i > 0)
			appendStringInfoString(&fc_buf, " AND ");

		appendStringInfoString(&fc_buf,
							   fc_quote_ident_cstr(NameStr(fc_attr->attname)));

		if (fc_tgt_pkattvals[fc_i] != NULL)
			appendStringInfo(&fc_buf, " = %s",
							 quote_literal_cstr(fc_tgt_pkattvals[fc_i]));
		else
			appendStringInfoString(&fc_buf, " IS NULL");
	}

	return fc_buf.data;
}

static char * fc_get_sql_update(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals, char **fc_tgt_pkattvals)
{
	char	   *fc_relname;
	HeapTuple	fc_tuple;
	TupleDesc	fc_tupdesc;
	int			fc_natts;
	StringInfoData fc_buf;
	char	   *fc_val;
	int			fc_key;
	int			fc_i;
	bool		fc_needComma;

	initStringInfo(&fc_buf);

	/* 获取关系名称，包括所需的模式前缀和引用 */
	fc_relname = fc_generate_relation_name(fc_rel);

	fc_tupdesc = fc_rel->rd_att;
	fc_natts = fc_tupdesc->natts;

	fc_tuple = fc_get_tuple_of_interest(fc_rel, fc_pkattnums, fc_pknumatts, fc_src_pkattvals);
	if (!fc_tuple)
		ereport(ERROR,
				(errcode(ERRCODE_CARDINALITY_VIOLATION),
				 errmsg("source row not found")));

	appendStringInfo(&fc_buf, "UPDATE %s SET ", fc_relname);

	/*
	 * 注意：i 是物理列号（从 0 开始计数）。
	 */
	fc_needComma = false;
	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_i);

		if (fc_attr->attisdropped)
			continue;

		if (fc_needComma)
			appendStringInfoString(&fc_buf, ", ");

		appendStringInfo(&fc_buf, "%s = ",
						 fc_quote_ident_cstr(NameStr(fc_attr->attname)));

		fc_key = fc_get_attnum_pk_pos(fc_pkattnums, fc_pknumatts, fc_i);

		if (fc_key >= 0)
			fc_val = fc_tgt_pkattvals[fc_key] ? pstrdup(fc_tgt_pkattvals[fc_key]) : NULL;
		else
			fc_val = SPI_getvalue(fc_tuple, fc_tupdesc, fc_i + 1);

		if (fc_val != NULL)
		{
			appendStringInfoString(&fc_buf, quote_literal_cstr(fc_val));
			pfree(fc_val);
		}
		else
			appendStringInfoString(&fc_buf, "NULL");
		fc_needComma = true;
	}

	appendStringInfoString(&fc_buf, " WHERE ");

	for (fc_i = 0; fc_i < fc_pknumatts; fc_i++)
	{
		int			fc_pkattnum = fc_pkattnums[fc_i];
		Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_pkattnum);

		if (fc_i > 0)
			appendStringInfoString(&fc_buf, " AND ");

		appendStringInfoString(&fc_buf,
							   fc_quote_ident_cstr(NameStr(fc_attr->attname)));

		fc_val = fc_tgt_pkattvals[fc_i];

		if (fc_val != NULL)
			appendStringInfo(&fc_buf, " = %s", quote_literal_cstr(fc_val));
		else
			appendStringInfoString(&fc_buf, " IS NULL");
	}

	return fc_buf.data;
}

/*
 * 返回一个适当引用的标识符。
 * 使用引用函数 quote_ident 在 quote.c 中。
 */
static char * fc_quote_ident_cstr(char *fc_rawstr)
{
	text	   *fc_rawstr_text;
	text	   *fc_result_text;
	char	   *fc_result;

	fc_rawstr_text = cstring_to_text(fc_rawstr);
	fc_result_text = DatumGetTextPP(DirectFunctionCall1(quote_ident,
													 PointerGetDatum(fc_rawstr_text)));
	fc_result = text_to_cstring(fc_result_text);

	return fc_result;
}

static int fc_get_attnum_pk_pos(int *fc_pkattnums, int fc_pknumatts, int fc_key)
{
	int			fc_i;

	/*
	 * 不太可能是一个长列表，所以只需扫描该值
	 */
	for (fc_i = 0; fc_i < fc_pknumatts; fc_i++)
		if (fc_key == fc_pkattnums[fc_i])
			return fc_i;

	return -1;
}

static HeapTuple fc_get_tuple_of_interest(Relation fc_rel, int *fc_pkattnums, int fc_pknumatts, char **fc_src_pkattvals)
{
	char	   *fc_relname;
	TupleDesc	fc_tupdesc;
	int			fc_natts;
	StringInfoData fc_buf;
	int			fc_ret;
	HeapTuple	fc_tuple;
	int			fc_i;

	/*
	 * 连接到 SPI 管理器
	 */
	if ((fc_ret = SPI_connect()) < 0)
		/* 内部错误 */
		elog(ERROR, "SPI connect failure - returned %d", fc_ret);

	initStringInfo(&fc_buf);

	/* 获取关系名称，包括所需的模式前缀和引用 */
	fc_relname = fc_generate_relation_name(fc_rel);

	fc_tupdesc = fc_rel->rd_att;
	fc_natts = fc_tupdesc->natts;

	/*
	 * 构建 SQL 语句以查找感兴趣的元组，即与
	 * src_pkattvals 匹配的那个。我们以前在这里使用 "SELECT *"，但生成
	 * 一个匹配表的物理结构的结果元组更简单，
	 * 且对任何被删除的列使用 NULL。否则我们必须处理两种
	 * 不同的 tupdescs，一切都会非常混乱。
	 */
	appendStringInfoString(&fc_buf, "SELECT ");

	for (fc_i = 0; fc_i < fc_natts; fc_i++)
	{
		Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_i);

		if (fc_i > 0)
			appendStringInfoString(&fc_buf, ", ");

		if (fc_attr->attisdropped)
			appendStringInfoString(&fc_buf, "NULL");
		else
			appendStringInfoString(&fc_buf,
								   fc_quote_ident_cstr(NameStr(fc_attr->attname)));
	}

	appendStringInfo(&fc_buf, " FROM %s WHERE ", fc_relname);

	for (fc_i = 0; fc_i < fc_pknumatts; fc_i++)
	{
		int			fc_pkattnum = fc_pkattnums[fc_i];
		Form_pg_attribute fc_attr = TupleDescAttr(fc_tupdesc, fc_pkattnum);

		if (fc_i > 0)
			appendStringInfoString(&fc_buf, " AND ");

		appendStringInfoString(&fc_buf,
							   fc_quote_ident_cstr(NameStr(fc_attr->attname)));

		if (fc_src_pkattvals[fc_i] != NULL)
			appendStringInfo(&fc_buf, " = %s",
							 quote_literal_cstr(fc_src_pkattvals[fc_i]));
		else
			appendStringInfoString(&fc_buf, " IS NULL");
	}

	/*
	 * 检索所需的元组
	 */
	fc_ret = SPI_exec(fc_buf.data, 0);
	pfree(fc_buf.data);

	/*
	 * 只允许一个符合条件的元组
	 */
	if ((fc_ret == SPI_OK_SELECT) && (SPI_processed > 1))
		ereport(ERROR,
				(errcode(ERRCODE_CARDINALITY_VIOLATION),
				 errmsg("source criteria matched more than one record")));

	else if (fc_ret == SPI_OK_SELECT && SPI_processed == 1)
	{
		SPITupleTable *fc_tuptable = SPI_tuptable;

		fc_tuple = SPI_copytuple(fc_tuptable->vals[0]);
		SPI_finish();

		return fc_tuple;
	}
	else
	{
		/*
		 * 没有符合条件的元组
		 */
		SPI_finish();

		return NULL;
	}

	/*
	 * 从未到达，但保持编译器安静
	 */
	return NULL;
}

/*
 * 打开由 relname_text 指定的关系，获取指定类型的锁，
 * 验证我们拥有指定的权限。
 * 调用方完成后必须关闭该关系。
 */
static Relation fc_get_rel_from_relname(text *fc_relname_text, LOCKMODE fc_lockmode, AclMode fc_aclmode)
{
	RangeVar   *fc_relvar;
	Relation	fc_rel;
	AclResult	fc_aclresult;

	fc_relvar = makeRangeVarFromNameList(textToQualifiedNameList(fc_relname_text));
	fc_rel = table_openrv(fc_relvar, fc_lockmode);

	fc_aclresult = pg_class_aclcheck(RelationGetRelid(fc_rel), GetUserId(),
								  fc_aclmode);
	if (fc_aclresult != ACLCHECK_OK)
		aclcheck_error(fc_aclresult, get_relkind_objtype(fc_rel->rd_rel->relkind),
					   RelationGetRelationName(fc_rel));

	return fc_rel;
}

/*
 * generate_relation_name - copied from ruleutils.c
 *		计算要显示的关系名称
 *
 * 结果包括所有必要的引号和模式前缀。
 */
static char * fc_generate_relation_name(Relation fc_rel)
{
	char	   *fc_nspname;
	char	   *fc_result;

	/* 如果在搜索路径中不可见，则限定名称 */
	if (RelationIsVisible(RelationGetRelid(fc_rel)))
		fc_nspname = NULL;
	else
		fc_nspname = get_namespace_name(fc_rel->rd_rel->relnamespace);

	fc_result = quote_qualified_identifier(fc_nspname, RelationGetRelationName(fc_rel));

	return fc_result;
}


static remoteConn *
fc_getConnectionByName(const char *fc_name)
{
	remoteConnHashEnt *fc_hentry;
	char	   *fc_key;

	if (!remoteConnHash)
		remoteConnHash = fc_createConnHash();

	fc_key = pstrdup(fc_name);
	truncate_identifier(fc_key, strlen(fc_key), false);
	fc_hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
											   fc_key, HASH_FIND, NULL);

	if (fc_hentry)
		return fc_hentry->rconn;

	return NULL;
}

static HTAB * fc_createConnHash(void)
{
	HASHCTL		fc_ctl;

	fc_ctl.keysize = NAMEDATALEN;
	fc_ctl.entrysize = sizeof(remoteConnHashEnt);

	return hash_create("Remote Con hash", NUMCONN, &fc_ctl,
					   HASH_ELEM | HASH_STRINGS);
}

static void fc_createNewConnection(const char *fc_name, remoteConn *fc_rconn)
{
	remoteConnHashEnt *fc_hentry;
	bool		fc_found;
	char	   *fc_key;

	if (!remoteConnHash)
		remoteConnHash = fc_createConnHash();

	fc_key = pstrdup(fc_name);
	truncate_identifier(fc_key, strlen(fc_key), true);
	fc_hentry = (remoteConnHashEnt *) hash_search(remoteConnHash, fc_key,
											   HASH_ENTER, &fc_found);

	if (fc_found)
	{
		PQfinish(fc_rconn->conn);
		ReleaseExternalFD();
		pfree(fc_rconn);

		ereport(ERROR,
				(errcode(ERRCODE_DUPLICATE_OBJECT),
				 errmsg("duplicate connection name")));
	}

	fc_hentry->rconn = fc_rconn;
	strlcpy(fc_hentry->name, fc_name, sizeof(fc_hentry->name));
}

static void fc_deleteConnection(const char *fc_name)
{
	remoteConnHashEnt *fc_hentry;
	bool		fc_found;
	char	   *fc_key;

	if (!remoteConnHash)
		remoteConnHash = fc_createConnHash();

	fc_key = pstrdup(fc_name);
	truncate_identifier(fc_key, strlen(fc_key), false);
	fc_hentry = (remoteConnHashEnt *) hash_search(remoteConnHash,
											   fc_key, HASH_REMOVE, &fc_found);

	if (!fc_hentry)
		ereport(ERROR,
				(errcode(ERRCODE_UNDEFINED_OBJECT),
				 errmsg("undefined connection name")));
}

static void fc_dblink_security_check(PGconn *fc_conn, remoteConn *fc_rconn)
{
	if (!superuser())
	{
		if (!PQconnectionUsedPassword(fc_conn))
		{
			PQfinish(fc_conn);
			ReleaseExternalFD();
			if (fc_rconn)
				pfree(fc_rconn);

			ereport(ERROR,
					(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
					 errmsg("password is required"),
					 errdetail("Non-superuser cannot connect if the server does not request a password."),
					 errhint("Target server's authentication method must be changed.")));
		}
	}
}

/*
 * 对于非超级用户，坚决要求 connstr 指定密码。 这
 * 防止从 .pgpass、服务文件、环境等中获取密码。 我们不希望
 * postgres 用户的密码对非超级用户可访问。
 */
static void fc_dblink_connstr_check(const char *fc_connstr)
{
	if (!superuser())
	{
		PQconninfoOption *fc_options;
		PQconninfoOption *fc_option;
		bool		fc_connstr_gives_password = false;

		fc_options = PQconninfoParse(fc_connstr, NULL);
		if (fc_options)
		{
			for (fc_option = fc_options; fc_option->keyword != NULL; fc_option++)
			{
				if (strcmp(fc_option->keyword, "password") == 0)
				{
					if (fc_option->val != NULL && fc_option->val[0] != '\0')
					{
						fc_connstr_gives_password = true;
						break;
					}
				}
			}
			PQconninfoFree(fc_options);
		}

		if (!fc_connstr_gives_password)
			ereport(ERROR,
					(errcode(ERRCODE_S_R_E_PROHIBITED_SQL_STATEMENT_ATTEMPTED),
					 errmsg("password is required"),
					 errdetail("Non-superusers must provide a password in the connection string.")));
	}
}

/*
 * 报告从远程服务器收到的错误
 *
 * res: 接收到的错误结果（将被释放）
 * fail: 对于 ERROR ereport 为 true，false 为 NOTICE
 * fmt 和后续参数: sprintf 风格的格式和 errcontext 的值;
 * 结果字符串应表述为“在 <某些操作> 时”
 */
static void fc_dblink_res_error(PGconn *fc_conn, const char *fc_conname, PGresult *fc_res,
				 bool fc_fail, const char *fc_fmt,...)
{
	int			fc_level;
	char	   *fc_pg_diag_sqlstate = PQresultErrorField(fc_res, PG_DIAG_SQLSTATE);
	char	   *fc_pg_diag_message_primary = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_PRIMARY);
	char	   *fc_pg_diag_message_detail = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_DETAIL);
	char	   *fc_pg_diag_message_hint = PQresultErrorField(fc_res, PG_DIAG_MESSAGE_HINT);
	char	   *fc_pg_diag_context = PQresultErrorField(fc_res, PG_DIAG_CONTEXT);
	int			fc_sqlstate;
	char	   *fc_message_primary;
	char	   *fc_message_detail;
	char	   *fc_message_hint;
	char	   *fc_message_context;
	va_list		fc_ap;
	char		fc_dblink_context_msg[512];

	if (fc_fail)
		fc_level = ERROR;
	else
		fc_level = NOTICE;

	if (fc_pg_diag_sqlstate)
		fc_sqlstate = MAKE_SQLSTATE(fc_pg_diag_sqlstate[0],
								 fc_pg_diag_sqlstate[1],
								 fc_pg_diag_sqlstate[2],
								 fc_pg_diag_sqlstate[3],
								 fc_pg_diag_sqlstate[4]);
	else
		fc_sqlstate = ERRCODE_CONNECTION_FAILURE;

	fc_message_primary = fc_xpstrdup(fc_pg_diag_message_primary);
	fc_message_detail = fc_xpstrdup(fc_pg_diag_message_detail);
	fc_message_hint = fc_xpstrdup(fc_pg_diag_message_hint);
	fc_message_context = fc_xpstrdup(fc_pg_diag_context);

	/*
	 * 如果我们没有从 PGresult 获得消息，请尝试 PGconn。 这是
	 * 必需的，因为对于连接级别的故障，PQexec 可能只返回
	 * NULL，而根本没有 PGresult。
	 */
	if (fc_message_primary == NULL)
		fc_message_primary = pchomp(PQerrorMessage(fc_conn));

	/*
	 * 由于我们已经复制了 PGresult 中所需的所有数据，因此
	 * 可以安全地释放它。 我们必须这样做以避免 PGresult 泄漏。 我们
	 * 也在泄漏所有字符串，但这些字符串是在 palloc 的内存中，最终会
	 * 被清理。
	 */
	if (fc_res)
		PQclear(fc_res);

	/*
	 * 格式化基本的 errcontext 字符串。 下面，我们将添加一些
	 * 关于连接名称的内容。 这是违反了将错误消息构造部分的可翻译性
	 * 指南，但由于 dblink 没有翻译支持，因此不需要担心
	 * (目前)。
	 */
	va_start(fc_ap, fc_fmt);
	vsnprintf(fc_dblink_context_msg, sizeof(fc_dblink_context_msg), fc_fmt, fc_ap);
	va_end(fc_ap);

	ereport(fc_level,
			(errcode(fc_sqlstate),
			 (fc_message_primary != NULL && fc_message_primary[0] != '\0') ?
			 errmsg_internal("%s", fc_message_primary) :
			 errmsg("could not obtain message string for remote error"),
			 fc_message_detail ? errdetail_internal("%s", fc_message_detail) : 0,
			 fc_message_hint ? errhint("%s", fc_message_hint) : 0,
			 fc_message_context ? (errcontext("%s", fc_message_context)) : 0,
			 fc_conname ?
			 (errcontext("%s on dblink connection named \"%s\"",
						 fc_dblink_context_msg, fc_conname)) :
			 (errcontext("%s on unnamed dblink connection",
						 fc_dblink_context_msg))));
}

/*
 * 获取外部服务器的连接字符串
 */
static char * fc_get_connect_string(const char *fc_servername)
{
	ForeignServer *fc_foreign_server = NULL;
	UserMapping *fc_user_mapping;
	ListCell   *fc_cell;
	StringInfoData fc_buf;
	ForeignDataWrapper *fc_fdw;
	AclResult	fc_aclresult;
	char	   *fc_srvname;

	static const PQconninfoOption *fc_options = NULL;

	initStringInfo(&fc_buf);

	/*
	 * 获取有效的 libpq 选项列表。
	 *
	 * 为了避免不必要的工作，我们只获取一次列表，并在
	 * 本后端进程的生命周期内使用它。我们无需担心
	 * 内存上下文问题，因为 PQconndefaults 使用 malloc 分配。
	 */
	if (!fc_options)
	{
		fc_options = PQconndefaults();
		if (!fc_options)			/* 假设失败的原因是 OOM */
			ereport(ERROR,
					(errcode(ERRCODE_FDW_OUT_OF_MEMORY),
					 errmsg("out of memory"),
					 errdetail("Could not get libpq's default connection options.")));
	}

	/* 首先收集服务器 connstr 选项 */
	fc_srvname = pstrdup(fc_servername);
	truncate_identifier(fc_srvname, strlen(fc_srvname), false);
	fc_foreign_server = GetForeignServerByName(fc_srvname, true);

	if (fc_foreign_server)
	{
		Oid			fc_serverid = fc_foreign_server->serverid;
		Oid			fc_fdwid = fc_foreign_server->fdwid;
		Oid			fc_userid = GetUserId();

		fc_user_mapping = GetUserMapping(fc_userid, fc_serverid);
		fc_fdw = GetForeignDataWrapper(fc_fdwid);

		/* 检查权限，用户必须在服务器上具有使用权限。 */
		fc_aclresult = pg_foreign_server_aclcheck(fc_serverid, fc_userid, ACL_USAGE);
		if (fc_aclresult != ACLCHECK_OK)
			aclcheck_error(fc_aclresult, OBJECT_FOREIGN_SERVER, fc_foreign_server->servername);

		foreach(fc_cell, fc_fdw->options)
		{
			DefElem    *fc_def = lfirst(fc_cell);

			if (fc_is_valid_dblink_option(fc_options, fc_def->defname, ForeignDataWrapperRelationId))
				appendStringInfo(&fc_buf, "%s='%s' ", fc_def->defname,
								 fc_escape_param_str(strVal(fc_def->arg)));
		}

		foreach(fc_cell, fc_foreign_server->options)
		{
			DefElem    *fc_def = lfirst(fc_cell);

			if (fc_is_valid_dblink_option(fc_options, fc_def->defname, ForeignServerRelationId))
				appendStringInfo(&fc_buf, "%s='%s' ", fc_def->defname,
								 fc_escape_param_str(strVal(fc_def->arg)));
		}

		foreach(fc_cell, fc_user_mapping->options)
		{

			DefElem    *fc_def = lfirst(fc_cell);

			if (fc_is_valid_dblink_option(fc_options, fc_def->defname, UserMappingRelationId))
				appendStringInfo(&fc_buf, "%s='%s' ", fc_def->defname,
								 fc_escape_param_str(strVal(fc_def->arg)));
		}

		return fc_buf.data;
	}
	else
		return NULL;
}

/*
 * 转义 libpq 连接参数字符串。
 *
 * 将 "'" 替换为 "\'" 和 "\" 替换为 "\\"。
 */
static char * fc_escape_param_str(const char *fc_str)
{
	const char *fc_cp;
	StringInfoData fc_buf;

	initStringInfo(&fc_buf);

	for (fc_cp = fc_str; *fc_cp; fc_cp++)
	{
		if (*fc_cp == '\\' || *fc_cp == '\'')
			appendStringInfoChar(&fc_buf, '\\');
		appendStringInfoChar(&fc_buf, *fc_cp);
	}

	return fc_buf.data;
}

/*
 * 验证 dblink_build_sql_insert() 和相关
 * 函数的 PK-attnums 参数，并转换为内部表示。
 *
 * 用户提供了一个基于 1 的逻辑 attnums 的 int2vector，以及一个计数
 * 参数（单独的计数参数的需要是历史遗留，但我们仍然检查它）。 我们检查每个
 * attnum 是否对应于 rel 的有效、未删除的属性。 我们
 * 不防止 attnums 被列出两次，尽管这种情况的实际用例是可疑的。
 * 请注意，在 Postgres 9.0 之前，用户的 attnums 被解释为
 * 物理而非逻辑列号；这已更改以确保未来的兼容性。
 *
 * 内部表示是一个 palloc 的 0 基数物理 attnums 的整数数组。
 */
static void fc_validate_pkattnums(Relation fc_rel,
				   int2vector *fc_pkattnums_arg, int32 fc_pknumatts_arg,
				   int **fc_pkattnums, int *fc_pknumatts)
{
	TupleDesc	fc_tupdesc = fc_rel->rd_att;
	int			fc_natts = fc_tupdesc->natts;
	int			fc_i;

	/* 不要获取超过数组元素数量的元素 */
	fc_pknumatts_arg = Min(fc_pknumatts_arg, fc_pkattnums_arg->dim1);

	/* 必须选择至少一个 pk attnum */
	if (fc_pknumatts_arg <= 0)
		ereport(ERROR,
				(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
				 errmsg("number of key attributes must be > 0")));

	/* 分配输出数组 */
	*fc_pkattnums = (int *) palloc(fc_pknumatts_arg * sizeof(int));
	*fc_pknumatts = fc_pknumatts_arg;

	/* 验证 attnums 并转换为内部形式 */
	for (fc_i = 0; fc_i < fc_pknumatts_arg; fc_i++)
	{
		int			fc_pkattnum = fc_pkattnums_arg->values[fc_i];
		int			fc_lnum;
		int			fc_j;

		/* 如果超出范围，可以立即抛出错误 */
		if (fc_pkattnum <= 0 || fc_pkattnum > fc_natts)
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid attribute number %d", fc_pkattnum)));

		/* 确定哪个物理列具有此逻辑编号 */
		fc_lnum = 0;
		for (fc_j = 0; fc_j < fc_natts; fc_j++)
		{
			/* 被丢弃的列不计算 */
			if (TupleDescAttr(fc_tupdesc, fc_j)->attisdropped)
				continue;

			if (++fc_lnum == fc_pkattnum)
				break;
		}

		if (fc_j < fc_natts)
			(*fc_pkattnums)[fc_i] = fc_j;
		else
			ereport(ERROR,
					(errcode(ERRCODE_INVALID_PARAMETER_VALUE),
					 errmsg("invalid attribute number %d", fc_pkattnum)));
	}
}

/*
 * 检查指定的连接选项是否有效。
 *
 * 我们基本上允许 libpq 认为是一个选项的任何内容，限制如下：
 *     调试选项：不允许
 *     "client_encoding"：不允许
 *     "user"：仅在 USER MAPPING 选项中有效
 *     安全选项（例如密码）：仅在 USER MAPPING 选项中有效
 *     其他：仅在 FOREIGN SERVER 选项中有效
 *
 * 我们不允许 client_encoding，因为它无论如何会通过
 * PQclientEncoding 被覆盖；允许指定它只会引起
 * 混淆。
 */
static bool fc_is_valid_dblink_option(const PQconninfoOption *fc_options, const char *fc_option,
					   Oid fc_context)
{
	const PQconninfoOption *fc_opt;

	/* 在 libpq 结果中查找选项 */
	for (fc_opt = fc_options; fc_opt->keyword; fc_opt++)
	{
		if (strcmp(fc_opt->keyword, fc_option) == 0)
			break;
	}
	if (fc_opt->keyword == NULL)
		return false;

	/* 不允许调试选项（尤其是 "replication"） */
	if (strchr(fc_opt->dispchar, 'D'))
		return false;

	/* 不允许 "client_encoding" */
	if (strcmp(fc_opt->keyword, "client_encoding") == 0)
		return false;

	/*
	 * 如果选项是 "user" 或标记为安全，则应仅在
	 * USER MAPPING 中指定。其他应仅在 SERVER 中指定。
	 */
	if (strcmp(fc_opt->keyword, "user") == 0 || strchr(fc_opt->dispchar, '*'))
	{
		if (fc_context != UserMappingRelationId)
			return false;
	}
	else
	{
		if (fc_context != ForeignServerRelationId)
			return false;
	}

	return true;
}

/*
 * 复制影响数据类型 I/O 的远程会话的 GUC 值
 * 并在新的 GUC 嵌套级别中应用这些值。返回新的
 * nestlevel（restoreLocalGucs 需要它以撤销设置），
 * 如果不需要新 nestlevel，则返回 -1。
 *
 * 我们使用相当于 SET 选项的函数，以允许设置
 * 仅在调用者调用 restoreLocalGucs 之前持续存在。如果
 * 期间抛出错误，guc.c 会处理撤销设置。
 */
static int fc_applyRemoteGucs(PGconn *fc_conn)
{
	static const char *const fc_GUCsAffectingIO[] = {
		"DateStyle",
		"IntervalStyle"
	};

	int			fc_nestlevel = -1;
	int			fc_i;

	for (fc_i = 0; fc_i < lengthof(fc_GUCsAffectingIO); fc_i++)
	{
		const char *fc_gucName = fc_GUCsAffectingIO[fc_i];
		const char *fc_remoteVal = PQparameterStatus(fc_conn, fc_gucName);
		const char *fc_localVal;

		/*
		 * 如果远程服务器是 8.4 版本之前，它将没有 IntervalStyle，
		 * 但这没关系，因为它的输出格式不会模糊。因此，如果我们
		 * 没有获得其值，就跳过该 GUC。（我们可能最终需要更复杂的
		 * 逻辑来检查远程版本。）
		 */
		if (fc_remoteVal == NULL)
			continue;

		/*
		 * 如果远程和本地 GUC 已经具有相同的值，避免 GUC 设置的开销。
		 */
		fc_localVal = GetConfigOption(fc_gucName, false, false);
		Assert(fc_localVal != NULL);

		if (strcmp(fc_remoteVal, fc_localVal) == 0)
			continue;

		/* 如果我们尚未创建新的 GUC 嵌套级别，则创建新的 */
		if (fc_nestlevel < 0)
			fc_nestlevel = NewGUCNestLevel();

		/* 应用选项（如果失败将抛出错误） */
		(void) set_config_option(fc_gucName, fc_remoteVal,
								 PGC_USERSET, PGC_S_SESSION,
								 GUC_ACTION_SAVE, true, 0, false);
	}

	return fc_nestlevel;
}

/*
 * 在它们被远程设置覆盖后，恢复本地 GUC。
 */
static void fc_restoreLocalGucs(int fc_nestlevel)
{
	/* 如果没有创建新的 nestlevel，则不执行任何操作 */
	if (fc_nestlevel > 0)
		AtEOXact_GUC(true, fc_nestlevel);
}
